package com.twoandone.tms.realtime.app.dwd;


import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class DwdServiceTaskCollect {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment STE = StreamTableEnvironment.create(env);

        STE.executeSql(KafkaUtil.getTopicDbDDL("Dwd_Service_express_task_collect1"));

        Table expressCollect = STE.sqlQuery(" select " +
                "        `data`['id'] id,\n" +
                "        `data`['order_id'] order_id,\n" +
                "        `data`['status'] status ,\n" +
                "        `data`['org_id'] org_id,\n" +
                "        `data`['org_name'] org_name,\n" +
                "        `data`['courier_emp_id'] courier_emp_id,\n" +
                "        `data`['courier_name'] courier_name,\n" +
                "        `data`['estimated_collected_time'] estimated_collected_time,\n" +
                "        `data`['estimated_commit_time'] estimated_commit_time,\n" +
                "        `data`['actual_collected_time'] actual_collected_time,\n" +
                "        `data`['actual_commit_time'] actual_commit_time,\n" +
                "        `data`['cancel_time'] cancel_time,\n" +
                "        `data`['create_time'] create_time,\n" +
                "        `data`['update_time'] update_time,\n" +
                "        `data`['is_deleted'] is_deleted,\n" +
                "        proc_time,\n" +
                "        ts\n" +
                "   from topic_db\n" +
                "   where `table` = 'express_task_collect'\n" +
                "   and `type` = 'insert' or  `type`  =  'update' ");
        STE.createTemporaryView("expressCollect", expressCollect);
        // STE.executeSql("select * from expressCollect").print();

        // 建立字典表
        STE.executeSql(MySqlUtil.getBaseDicLookUpDDL());
        // // STE.executeSql("select * from base_dic").print();


        Table lookUpJoin = STE.sqlQuery(" SELECT \n" +
                "            `Pick`.id,\n" +
                "            `Pick`.order_id,\n" +
                "             `Pick`.status,\n" +
                "            `Pick`.org_id,\n" +
                "            `Pick`.org_name,\n" +
                "            `Pick`.courier_emp_id,\n" +
                "            `Pick`.courier_name,\n" +
                "            `Pick`.estimated_collected_time,\n" +
                "            `Pick`.estimated_commit_time,\n" +
                "            `Pick`.actual_collected_time,\n" +
                "            `Pick`.actual_commit_time,\n" +
                "            `Pick`.cancel_time,\n" +
                "            `Pick`.create_time,\n" +
                "            `Pick`.update_time,\n" +
                "            `Pick`.is_deleted,\n" +
                "            dic.name status_name\n" +//CAST(`Pick`.status AS String)
                "           FROM expressCollect AS `Pick` JOIN base_dic FOR SYSTEM_TIME AS OF `Pick`.proc_time AS dic ON CAST(`Pick`.status AS bigint) = dic.id");
        STE.createTemporaryView("lookUpJoin", lookUpJoin);
        // STE.executeSql("select * from lookUpJoin").print();
        // //创建动态表和kafka主题映射
        STE.executeSql("CREATE TABLE dwd_service_express_collect (\n" +
                "              id string,\n" +
                "              order_id string,\n" +
                "              status string,\n" +
                "              org_id string,\n" +
                "              org_name string,\n" +
                "              courier_emp_id string,\n" +
                "              courier_name string,\n" +
                "              estimated_collected_time string,\n" +
                "              estimated_commit_time string,\n" +
                "              actual_collected_time string,\n" +
                "              actual_commit_time string,\n" +
                "              cancel_time string,\n" +
                "              create_time string,\n" +
                "              update_time string,\n" +
                "              is_deleted string,\n" +
                "              status_name string,\n" +
                "              PRIMARY KEY (id) NOT ENFORCED )\n"
                + KafkaUtil.getUpsertKafkaDDL("dwd_service_express_collect"));

        //写入kafka
        STE.executeSql("insert into dwd_service_express_collect select * from lookUpJoin");


    }
}
